package org.activequant.data.util.subscription;

import java.util.ArrayList;
import java.util.List;

import org.activequant.core.domainmodel.InstrumentSpecification;
import org.activequant.core.domainmodel.data.Quote;
import org.activequant.core.types.TimeFrame;
import org.activequant.data.retrieval.IQuoteSubscriptionSource;
import org.activequant.data.retrieval.ISubscription;
import org.activequant.data.retrieval.SubscriptionSourceBase;
import org.activequant.util.pattern.events.IEventListener;

/**
 * Class that merges trade indication event streams coming from several
 * subscription sources.
 * When a subscription request is received, this class will open a subscription at
 * every configured source, events coming from all sources are merged and 
 * sent to the user.<br/>
 * QuoteMergingSubscriptionSource extends SubscriptionSourceBase&lt;Quote&gt; implements IQuoteSubscriptionSource.<br/>
 * Defines a nested MultiSubscription class. Holds the following associated variables:
 * <ul>
 * <li>sources(IQuoteSubscriptionSource[])</li>
 * </ul>
 * Holds the following inherited variables:
 * <ul>
 * <li>activeBackends(Map&lt;Topic, Subscription&gt;)</li>
 * <li>activeSubscriptions(Queue&lt;SubscriptionDelegate&gt;)</li>
 * </ul>
 * <b>History:</b><br>
 *  - [11.11.2007] Created (Mike Kroutikov)<br>
 *
 *  @author Mike Kroutikov
 */
public class QuoteMergingSubscriptionSource extends SubscriptionSourceBase<Quote> implements IQuoteSubscriptionSource {
	/**
	 * An empty QuoteMergingSubscriptionSource (extends SubscriptionSourceBase&lt;Quote&gt; implements IQuoteSubscriptionSource) constructor
	 */
	public QuoteMergingSubscriptionSource() {
		super();
	}
	/**
	 * private IQuoteSubscriptionSource[] sources;
	 */
	private IQuoteSubscriptionSource[] sources;
	/**
	 * returns the associated sources(IQuoteSubscriptionSource[])
	 * @return
	 */
	public IQuoteSubscriptionSource [] getSources() { 
		return sources;
	}
	/**
	 * sets the associated sources(IQuoteSubscriptionSource[]) with the given val(IQuoteSubscriptionSource[])
	 * @param val
	 */
	public void setSources(IQuoteSubscriptionSource [] val) {
		sources = val;
	}
	/**
	 * A private class. MultiSubscription extends Subscription implements IEventListener&lt;Quote&gt;.<br/>
	 * Holds the following associated variables:
	 * <ul>
	 * <li>subscriptions(List&lt;ISubscription&lt;Quote&gt;&gt;)</li>
	 * </ul>
	 * Holds the following inherited variables:
	 * <ul>
	 * <li>isActive(AtomicBoolean)</li>
	 * <li>refCount(AtomicInteger)</li>
	 * <li>instrument(InstrumentSpecification)</li>
	 * <li>event(Event&lt;Quote&gt;)</li>
	 * </ul>
	 * @author Dimitar
	 *
	 */
	private class MultiSubscription extends Subscription implements IEventListener<Quote> {
		/**
		 * private final List&lt;ISubscription&lt;Quote&gt;&gt; subscriptions = new ArrayList&lt;ISubscription&lt;Quote&gt;&gt;();
		 */
		private final List<ISubscription<Quote>> subscriptions = new ArrayList<ISubscription<Quote>>();
		/**
		 * Constructs a MultiSubscription (extends Subscription implements IEventListener&lt;Quote&gt;).<br/>
		 * Creates an ISubscription&lt;Quote&gt; using each one of the sources(IQuoteSubscriptionSource[]) for the given spec(InstrumentSpecification)
		 * and adds them to the associated subscriptions(List&lt;ISubscription&lt;Quote&gt;&gt;). It also registers this MultiSubscription (implements IEventListener&lt;Quote&gt;)
		 * instance to listen to events(Quote) from each of those subscriptions.
		 * @param spec
		 * @throws Exception
		 */
		public MultiSubscription(InstrumentSpecification spec) throws Exception {
			for(IQuoteSubscriptionSource source : sources) {
				ISubscription<Quote> s = source.subscribe(spec); 
				subscriptions.add(s);
				s.addEventListener(this);
			}
		}
		/**
		 * Loops through the associated subscriptions(List&lt;ISubscription&lt;Quote&gt;&gt;) and activates each one of them.
		 */
		@Override
		protected void handleActivate() throws Exception {
			for(ISubscription<Quote> s : subscriptions) {
				s.activate();
			}
		}
		/**
		 * Loops through the associated subscriptions(List&lt;ISubscription&lt;Quote&gt;&gt;) and stops itself from listening to events from them and cancels each one of them.
		 */
		@Override
		protected void handleCancel() throws Exception {
			for(ISubscription<Quote> s : subscriptions) {
				s.removeEventListener(this);
				s.cancel();
			}
		}
		/**
		 * invokes <code>event.fire(event)</code> on the inherited event(Event&lt;Quote&gt;) using the given event(Quote)
		 */
		public void eventFired(Quote event) {
			fireEvent(event);
		}
	}
	/**
	 * Creates and returns a MultiSubscription (extends Subscription implements IEventListener&lt;Quote&gt;).<br/>
	 * Creates an ISubscription&lt;Quote&gt; using each one of the associated sources(IQuoteSubscriptionSource[]) for the <strong>given spec(InstrumentSpecification)</strong>
	 * and adds them to its associated subscriptions(List&lt;ISubscription&lt;Quote&gt;&gt;). It also registers this MultiSubscription (implements IEventListener&lt;Quote&gt;)
	 * instance to listen to events(Quote) from each of those subscriptions.
	 */
	@Override
	protected Subscription createSubscription(InstrumentSpecification spec,
			TimeFrame timeFrame) {
		try {
			return new MultiSubscription(spec);
		} catch (Exception e) {
			throw new RuntimeException(e);
		}
	}
	/**
	 * Returns/creates a SubscriptionDelegate(implements ISubscription&lt;Quote&gt;) for the given spec(InstrumentSpecification) and TIMEFRAME_1_TICK.
	 */
	public ISubscription<Quote> subscribe(InstrumentSpecification spec)
			throws Exception {
		return super.subscribe(spec, TimeFrame.TIMEFRAME_1_TICK);
	}
}
